package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeRefundPaySucDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeRefundPaySucDetail().start(10018,4, Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1. 读取 topic_db
        readOdsDb(tEnv,groupId);

        // 2. 读取 字典表
        createBaseDic(tEnv);

        // 3. 过滤退款成功表数据
        refund_paymentStream(tEnv);

        // 4. 过滤退单表中的退单成功的数据
        refund_infoStream(tEnv);

        // 5. 过滤订单表中的退款成功的数据
        order_info(tEnv);
        // 6. 4 张表的 join
        Table table = four_joinStream(tEnv);

        // 7.写出到 kafka

        //创建 kafka 框架
        create_kafka(tEnv);

        //插入数据
        table.insertInto(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS).execute();


    }

    private static void create_kafka(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS + "\n" +
                "(\n" +
                " id String,\n" +
                " user_id String,\n" +
                " order_id String,\n" +
                " sku_id String,\n" +
                " province_id String,\n" +
                " payment_type String,\n" +
                " payment_type_name String,\n" +
                " date_id String,\n" +
                " callback_time String,\n" +
                " refund_num String,\n" +
                " total_amount String,\n" +
                " ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS));
    }

    private static Table four_joinStream(StreamTableEnvironment tEnv) {
        return  tEnv.sqlQuery("select\n" +
                "rp.id,\n" +
                "oi.user_id,\n" +
                "rp.order_id,\n" +
                "rp.sku_id,\n" +
                "oi.province_id,\n" +
                "rp.payment_type,\n" +
                "dic.info.dic_name payment_type_name,\n" +
                "date_format(rp.callback_time,'yyyy-MM-dd') date_id,\n" +
                "rp.callback_time,\n" +
                "ori.refund_num,\n" +
                "rp.total_amount,\n" +
                "rp.ts\n" +
                "from refund_payment rp\n" +
                "join refund_info ori \n" +
                "on rp.order_id=ori.order_id and rp.sku_id =ori.sku_id\n" +
                "join order_info oi\n" +
                "on rp.order_id=oi.id\n" +
                "join base_dic for system_time as of rp.pt as dic\n" +
                "on rp.payment_type=dic.rowkey");
    }

    private static void order_info(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['province_id'] province_id\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='order_info'\n" +
                "and `type`='update'\n" +
                "and `old`['order_status'] is not null\n" +
                "and `data`['order_status']='1006'");

        tEnv.createTemporaryView("order_info",table);
    }

    private static void refund_infoStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['refund_num'] refund_num\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='order_refund_info'\n" +
                "and `type`='update'\n" +
                "and `old`['refund_status'] is not null\n" +
                "and `data`['refund_status']='0705'");

        tEnv.createTemporaryView("refund_info",table);
    }

    private static void refund_paymentStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['callback_time'] callback_time,\n" +
                "data['total_amount'] total_amount,\n" +
                "proctime() pt,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table`='refund_payment'\n" +
                "and `type`='update'\n" +
                "and `old`['refund_status'] is not null\n" +
                "and `data`['refund_status']='1602'");

        tEnv.createTemporaryView("refund_payment",table);
    }
}
